#include "Publisher_profile.h"
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
using namespace eprosima::fastdds::dds;
using namespace eprosima::fastdds::rtps;
using namespace eprosima::fastrtps::rtps;

std::string getPublishMsg(uint32_t send_size) {
	FILE* fp = nullptr;
	fp = fopen("/home/wang/Fast-DDS-Dev/dds/profile_example/publishData.txt", "rb");
	if (!fp) {
		printf("file open failed.\n");
		return "file open failed";
	}
	char* buf = (char*)malloc(send_size);
	size_t s = fread(buf, send_size, send_size, fp);
	std::string res(std::move(buf), s);
	fclose(fp);
	free(buf);
	printf("send size:%d\n", send_size);
	return res;
}
time_t getTimestamp() {
	std::chrono::time_point<std::chrono::system_clock, std::chrono::microseconds> tpMicro
		= std::chrono::time_point_cast<std::chrono::microseconds>(std::chrono::system_clock::now());
	time_t totalMicroSeconds = tpMicro.time_since_epoch().count();
	return totalMicroSeconds;
}
Publisher_profile::Publisher_profile():participant_(nullptr), type_(new MessagePubSubType())
{
	stop_.store(true);
}

Publisher_profile::~Publisher_profile()
{
}

bool Publisher_profile::init(std::string xml_path, std::string participant_profile, 
	std::string topic_profile, std::string datawriter_profile, std::string topic_name)
{
	if (publishers_.count(topic_name)) {
		return true;
	}
	if (!participant_) {
		auto ret = DomainParticipantFactory::get_instance()->load_XML_profiles_file(xml_path);
		if (ReturnCode_t::RETCODE_OK != ret)
		{
			return false;
		}
		participant_ =
			DomainParticipantFactory::get_instance()->create_participant_with_profile(12, participant_profile);
		if (!participant_) {
			return false;
		}
		type_.register_type(participant_);
	}
	Topic* topic = nullptr;
	Publisher* publisher = nullptr;
	DataWriter* datawriter = nullptr;
	do {
		topic =
			participant_->create_topic_with_profile(topic_name, "Message", topic_profile);
		if (!topic) {
			break;
		}

		publisher = participant_->create_publisher(PUBLISHER_QOS_DEFAULT);
		if (!publisher) {
			break;
		}
		auto listener = new PubListener();
		datawriter = publisher->create_datawriter_with_profile(topic, datawriter_profile, listener);
		if (!datawriter) {
			break;
		}
		auto writer_qos = datawriter->get_qos().reliability();
		auto ptp_qos = participant_->get_qos();
		//ptp_qos.transport().user_transports();
		publishers_[topic_name] = publisher;
		stop_.store(false);
		return true;
	} while (false);

	if (topic != nullptr) {
		participant_->delete_topic(topic);
	}
	if (datawriter != nullptr) {
		auto listener = publisher->get_listener();
		delete listener;
		publisher->delete_datawriter(datawriter);
	}
	if (publisher != nullptr) {
		participant_->delete_publisher(publisher);
	}

	return false;
}

bool Publisher_profile::publish(std::string topic_name, Message &msg)
{
	if (!publishers_.count(topic_name)) {
		std::cout << "topic name is not exist." << topic_name << std::endl;
		return false;
	}
	
	auto writer = publishers_[topic_name]->lookup_datawriter(topic_name);
	return writer->write(&msg);
}

void Publisher_profile::run(std::string topic_name, uint32_t samples, uint32_t sleep, uint32_t send_size)
{
	std::thread thread(&Publisher_profile::runThread, this, topic_name, samples, sleep, send_size);
	if (samples == 0)
	{
		std::cout << "Publisher running. Please press enter to stop the Publisher at any time." << std::endl;
		//std::cin.ignore();
	}
	else
	{
		std::cout << "Publisher running " << samples << " samples." << std::endl;
	}
	//thread.join();
	threads_[topic_name].swap(thread);
}

void Publisher_profile::stop()
{
	stop_.store(true);
}

void Publisher_profile::runThread(std::string topic_name, uint32_t samples, uint32_t sleep, uint32_t send_size)
{
	Message msg;
	msg.message(getPublishMsg(send_size));
	msg.index(0);
	for (uint32_t i = 0; (i < samples) && !stop_; ++i)
	{
		auto t = getTimestamp();
		msg.timestamp(t);
		msg.index(msg.index() + 1);
		publish(topic_name, msg);
		//auto diff = getTimestamp() - t;
		std::this_thread::sleep_for(std::chrono::microseconds(sleep));
	}
	printf("publisher finished. samples: %d\n", samples);
}
